package com.geekbang.work.kmq.v3.broker;

import com.geekbang.work.kmq.v3.KmqMessage;
import com.geekbang.work.kmq.v3.broker.core.KmqBroker;
import com.geekbang.work.kmq.v3.broker.core.KmqConsumer;
import com.geekbang.work.kmq.v3.broker.core.KmqProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author nhsoft.lul
 */
@SpringBootApplication
@RestController
public class BrokerApplication {
    public static void main(String args[]) {
        SpringApplication.run(BrokerApplication.class, args);
    }

    @Autowired
    private KmqBroker kmqBroker;

    @PutMapping(value = "send/{topic}",headers = "Content-Type=application/json")
    public Boolean send(@PathVariable("topic") String topic, @RequestBody KmqMessage kmqMessage){
        kmqBroker.createTopic(topic);
        KmqProducer producer = kmqBroker.createProducer();
        producer.send(topic, kmqMessage);
        return true;
    }

    @GetMapping(value = "poll/{topic}")
    public KmqMessage poll(@PathVariable("topic") String topic, @RequestParam(value = "offset", required = false) Integer offset){
        kmqBroker.createTopic(topic);
        KmqConsumer consumer = kmqBroker.createConsumer();
        consumer.subscribe(topic);
        KmqMessage kmqMessage = consumer.poll(offset);
        return kmqMessage;
    }
}
